package org.wu.framework.easy.upsert.sink;


import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.wu.framework.easy.upsert.autoconfigure.EasySmart;
import org.wu.framework.easy.upsert.autoconfigure.config.SpringUpsertAutoConfigure;
import org.wu.framework.easy.upsert.autoconfigure.dynamic.EasyUpsertStrategy;
import org.wu.framework.easy.upsert.autoconfigure.enums.EasyUpsertType;
import org.wu.framework.easy.upsert.autoconfigure.sink.LocalStorageClassAnnotation;
import org.wu.framework.easy.upsert.core.dynamic.IEasyUpsert;
import org.wu.framework.easy.upsert.core.dynamic.exception.UpsertException;
import org.wu.framework.easy.upsert.core.dynamic.function.EasyUpsertFunction;
import org.wu.framework.easy.upsert.sink.converter.ConverterClass2KafkaSchema;
import org.wu.framework.easy.upsert.sink.converter.JsonFileConverter;
import org.wu.framework.easy.upsert.sink.kafka.KafkaJsonMessage;
import org.wu.framework.easy.upsert.sink.kafka.TargetJsonSchema;
import org.wu.framework.lazy.orm.core.persistence.analyze.EasyAnnotationConverter;
import org.wu.framework.translation.data.ClassSchema;
import org.wu.framework.translation.data.IBeanUpsert;
import org.wu.framework.translation.data.ProcessException;
import org.wu.framework.translation.data.UserConvertService;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;

/**
 * description kafka数据插入
 *
 * @author Jia wei Wu
 * @date 2020/9/11 上午10:22
 */
@Slf4j
@ConditionalOnBean(EasyUpsertExtractKafkaProducer.class)
@EasyUpsertStrategy(value = EasyUpsertType.KAFKA)
public class KafkaEasyUpsertSink implements IEasyUpsert {

    private final UserConvertService userConvertService;
    private final SpringUpsertAutoConfigure springUpsertAutoConfigure;
    private final EasyUpsertExtractKafkaProducer easyUpsertExtractKafkaProducer;

    public KafkaEasyUpsertSink(UserConvertService userConvertService, SpringUpsertAutoConfigure springUpsertAutoConfigure, EasyUpsertExtractKafkaProducer easyUpsertExtractKafkaProducer) {
        this.userConvertService = userConvertService;
        this.springUpsertAutoConfigure = springUpsertAutoConfigure;
        this.easyUpsertExtractKafkaProducer = easyUpsertExtractKafkaProducer;
    }

    @Override
    public <T> Object upsert(List<T> list, ClassSchema classSchema) throws UpsertException, ExecutionException, InterruptedException {
        splitListThen(list, springUpsertAutoConfigure.getBatchLimit(), new EasyUpsertFunction() {
            @Override
            public <t> void handle(List<t> source) {
                Class<T> clazz = classSchema.clazz();
                // 模块名称+业务+表名
                EasySmart lazyTableAnnotation =
                        LocalStorageClassAnnotation.getEasyTableAnnotation(clazz, springUpsertAutoConfigure.isForceDuplicateNameSwitch());
                String schemaName = lazyTableAnnotation.kafkaSchemaName();

                TargetJsonSchema targetJsonSchema = KafkaJsonMessage.targetSchemaMap.get(schemaName);
                if (targetJsonSchema == null) {
                    synchronized (KafkaJsonMessage.targetSchemaMap) {
                        targetJsonSchema = ConverterClass2KafkaSchema.converterClass2TargetJsonSchema(clazz, springUpsertAutoConfigure.isForceDuplicateNameSwitch());
                        KafkaJsonMessage.targetSchemaMap = Maps.uniqueIndex(Arrays.asList(targetJsonSchema), TargetJsonSchema::getName);
                        log.info(" Automatic loading TargetJsonSchema for class {}", schemaName);
                    }
                }
                KafkaJsonMessage kafkaJsonMessage = KafkaJsonMessage.newInstance("", schemaName);

                Map iEnumList = new HashMap();
                if (null != userConvertService) {
                    iEnumList = userConvertService.userConvert(clazz);
                }
                iEnumList.putAll(EasyAnnotationConverter.collectionConvert(clazz));
                for (Object value : source) {
                    if (IBeanUpsert.class.isAssignableFrom(clazz)) {
                        try {
                            ((IBeanUpsert) value).beforeObjectProcess();
                        } catch (ProcessException e) {
                            e.printStackTrace();
                            throw new UpsertException(e);
                        }
                    }
                    kafkaJsonMessage.setPayload(JsonFileConverter.parseBean2map(value, iEnumList));
                    easyUpsertExtractKafkaProducer.sendAsync(lazyTableAnnotation.kafkaCode(),
                            lazyTableAnnotation.kafkaTopicName(), kafkaJsonMessage);
                }

            }
        });
        return true;
    }

}
